"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : 数据库系统
Case Name   : 人为修改配置文件中的审计目录为另一个无权限目录
Create At   : 2020/06/29
Owner       : @peilinqian
Description :
    1. 进行事务操作及登入登出操作，查看审计日志
    2. 人为修改配置文件中的审计目录为另一个无权限目录
    3. 重启数据库
    4. 恢复新目录权限
    chmod -R 700  /home/test_tmp
    5. 重启数据库生效
    6. 进行事务操作及登入登出操作，查看审计日志
    7. 恢复环境
Expect      :
    1、审计日志有1次登入登出记录
    2、修改成功
    3、重启成功
    4、恢复新目录权限成功
    5、重启成功
    6、审计日志有1次登入登出记录
    7、恢复环境
History     :
    modified : 2021/01/20 by @peilinqian 修改查看审计登录信息时间节点，以防主机与yat主机时间不一致导致查询失败
    modified : 2021/01/20 by @peilinqian 优化期望及teardown
    Modified by @li-xin12345 2024/1/22: 增加审计日志查看范围
    Modified by @li-xin12345 2024/2/26: 增加审计日志查看范围
"""

import datetime
import os
import random
import time
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.CommonSH import RestartDbCluster
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class ChangepgAudit(unittest.TestCase):
    sh = CommonSH()
    nodes_tuple = ('PrimaryDbUser', 'Standby1DbUser', 'Standby2DbUser')

    @RestartDbCluster(*nodes_tuple)
    def setUp(self):
        self.log = Logger()
        self.log.info('----Opengauss_Reliability_Dbsys_Case059.py start----')
        self.primary_root_node = Node('PrimaryRoot')
        self.primary_user_node = Node(node='PrimaryDbUser')
        self.com = Common()
        self.Constant = Constant()
        self.new_audit_path = os.path.join(macro.PG_AUDIT_PATH,
                                           macro.DN_NODE_NAME.split('/')[0],
                                           'reliability050')
        self.log.info(self.new_audit_path)
        self.pg_audit_path = os.path.join(macro.PG_AUDIT_PATH,
                                          macro.DN_NODE_NAME.split('/')[0])
        self.log.info(self.pg_audit_path)
        self.new_user = 'reliability050' + str(random.randint(1, 100))

        self.log.info('----获取主节点hostname----')
        cmd = "cat /etc/hostname"
        self.log.info(cmd)
        self.hostname = self.primary_root_node.sh(cmd).result()

        self.log.info("----打开审计开关，设置相关参数----")
        # 控制审计进程的开启和关闭。审计进程开启后，将从管道读取后台进程写入的审计信息，并写入审计文件。
        msg = self.sh.executeGsguc('reload', self.Constant.GSGUC_SUCCESS_MSG,
                                   'audit_enabled=on')
        self.assertTrue(msg)
        # 这个参数决定是否审计openGauss用户的登录（包括登录成功和登录失败）、注销
        msg = self.sh.executeGsguc('reload', self.Constant.GSGUC_SUCCESS_MSG,
                                   'audit_login_logout=7')
        self.assertTrue(msg)

        self.log.info("-----创建新用户------")
        sql_cmd = f"create user {self.new_user} " \
            f"with password '{macro.PASSWD_INITIAL}';"
        self.log.info(sql_cmd)
        msg = self.sh.executDbSql(sql_cmd)
        self.log.info(msg)
        for err_msg in self.Constant.SQL_WRONG_MSG:
            self.assertNotIn(err_msg, msg)

    def test_change_pgaudit(self):
        self.log.info("------修改审计目录前使用新用户登录-----")
        shell_cmd = f"source {macro.DB_ENV_PATH}; " \
            f"gsql -d {self.primary_user_node.db_name} " \
            f"-p {self.primary_user_node.db_port} " \
            f"-U '{self.new_user}' -W '{macro.PASSWD_INITIAL}' -l"
        self.log.info(shell_cmd)
        msg = self.primary_user_node.sh(shell_cmd).result()
        self.log.info(msg)
        for err_msg in self.Constant.SQL_WRONG_MSG:
            self.assertNotIn(err_msg, msg)

        time.sleep(90)

        self.log.info("查看审计记录")
        end_time = self.primary_root_node.sh(
            'date -d today +"%Y-%m-%d %H:%M:%S"').result()
        start_time = datetime.datetime.strptime(
            end_time, '%Y-%m-%d %H:%M:%S') - datetime.timedelta(minutes=6)

        query_audit_cmd = f"select count(*) from " \
            f"pg_query_audit('{start_time}','{end_time}') " \
            f"where type = 'login_success' and username = '{self.new_user}';"
        self.log.info(query_audit_cmd)
        query_result = self.sh.executDbSql(query_audit_cmd)
        self.log.info(query_result)
        result_dict = self.com.format_sql_result(query_result)
        self.assertEqual(int(result_dict.get('count')[0]), 1)

        self.log.info("-----创建目录并修改为无权限-----")
        cmd = f"mkdir {self.new_audit_path}; "
        self.log.info(cmd)
        msg = self.primary_user_node.sh(cmd).result()
        self.log.info(msg)
        self.log.info("----修改目录为无权限------")
        cmd = f"chmod 000 -R {self.new_audit_path}; " \
            f"ls -l {os.path.dirname(self.new_audit_path)}"
        self.log.info(cmd)
        msg = self.primary_root_node.sh(cmd).result()
        self.log.info(msg)

        self.log.info("----修改审计目录----")
        self.sh.executeGsguc('set', self.Constant.GSGUC_SUCCESS_MSG,
                             f"audit_directory='{self.new_audit_path}'",
                             self.hostname)

        self.log.info("-----重启成功--------")
        is_stopped = self.sh.stopDbCluster()
        self.assertTrue(is_stopped)
        is_started = self.sh.startDbCluster()
        self.assertTrue(is_started)

        self.log.info("----查看集群状态----")
        db_status = self.sh.getDbClusterStatus('detail')
        self.log.info(db_status)

        self.log.info("-----还原权限-----------")
        cmd = f"chmod 700 -R {self.new_audit_path}; ls -l " \
            f"{os.path.dirname(self.new_audit_path)}"
        self.log.info(cmd)
        msg = self.primary_root_node.sh(cmd).result()
        self.log.info(msg)

        self.log.info("-----重启生效--------")
        is_stopped = self.sh.stopDbCluster()
        self.assertTrue(is_stopped)
        is_started = self.sh.startDbCluster()
        self.assertTrue(is_started)

        self.log.info("----查看集群状态----")
        db_status = self.sh.getDbClusterStatus('detail')
        self.log.info(db_status)

        self.log.info("----修改审计目录后使用新用户登录----")
        shell_cmd = f"source {macro.DB_ENV_PATH}; " \
            f"gsql -d {self.primary_user_node.db_name} " \
            f"-p {self.primary_user_node.db_port} " \
            f"-U '{self.new_user}' " \
            f"-W '{macro.PASSWD_INITIAL}' -l"
        self.log.info(shell_cmd)
        msg = self.primary_user_node.sh(shell_cmd).result()
        self.log.info(msg)
        for err_msg in self.Constant.SQL_WRONG_MSG:
            self.assertNotIn(err_msg, msg)

        time.sleep(120)

        self.log.info("----查看审计记录---------")
        end_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        query_audit_cmd = f"select count(*) from pg_query_audit" \
            f"('{start_time}','{end_time}') where " \
            f"type = 'login_success' and username = '{self.new_user}';"
        self.log.info(query_audit_cmd)
        query_result = self.sh.executDbSql(query_audit_cmd)
        self.log.info(query_result)
        result_dict = self.com.format_sql_result(query_result)
        # 因修改了无权限目录，所以只能查到最近的一次记录
        self.assertEqual(int(result_dict.get('count')[0]), 1)

    def tearDown(self):
        self.log.info('this is tearDown')
        self.log.info("----删除新建用户----")
        sql_cmd = f"drop user {self.new_user};"
        self.log.info(sql_cmd)
        msg = self.sh.executDbSql(sql_cmd)
        self.log.info(msg)

        self.log.info("----删除创建目录----")
        cmd = f"rm -rf {self.new_audit_path}"
        self.log.info(cmd)
        msg = self.primary_root_node.sh(cmd).result()
        self.log.info(msg)

        self.log.info("----还原默认值----")
        # audit_enabled默认值为on，无需还原; audit_login_logout默认值为7，无需还原
        # audit_space_limit默认值为1GB，需还原
        # 还原审计目录
        self.sh.executeGsguc('set', self.Constant.GSGUC_SUCCESS_MSG,
                             f"audit_directory='{self.pg_audit_path}'",
                             self.hostname)

        self.log.info("重启生效")
        self.sh.stopDbCluster()
        self.sh.startDbCluster()